package com.bungarus.busi;

import com.bungarus.Busi;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.helper.RedisCacheHelper;
import com.bungarus.model.BungarusClientException;
import com.bungarus.model.MessageBuilder;
import com.bungarus.offline.OfflineMessagePersist;
import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import org.apache.log4j.Logger;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Set;

/**
 * Created by tang on 2019/1/1.
 */
public class GroupSender {
    Logger logger = Logger.getLogger(GroupSender.class);

    private static GroupSender groupSender = new GroupSender();

    private static final String PREFIX_GROUP = "groupmember:";
    private final int size = 200;

    private RedisClient groupMemberCache;
    private StatefulRedisConnection<String, String> connection;

    private GroupSender() {
        init();
    }

    public static GroupSender getInstance() {
        return groupSender;
    }

    private void init() {
        groupMemberCache = RedisCacheHelper.initRedisClient();
        connection = groupMemberCache.connect();
    }

    private StatefulRedisConnection<String, String> getConnection() {
        return connection;
    }

    public void destroy() {
        RedisCacheHelper.shutdownRedisClient(groupMemberCache);
    }

    public void transferToNodeProcessingLargeGroup(MessageProtobuf.Message msg, Busi busiServer) throws BungarusClientException {
        //transferToNodeProcessingLargeGroup
        List<InnerClient> bridgeClients = busiServer.getInnerClientList().get(Busi.CLUSTER_TAG_BUSI_BRIDGE);
        if(bridgeClients.size() > 0) {
            int index = Integer.parseInt(msg.getHeader().getId()) % bridgeClients.size();
            //prepare msg (add 'tag') for remedying
            bridgeClients.get(index).sendMessage(
                    MessageBuilder.buildBusiBridgeMsg(Busi.CLUSTER_TAG_BUSI_BRIDGE, msg),
                    Busi.CLUSTER_TAG_BUSI_BRIDGE);
        }
    }

    public void sendToGroup(String groupId, MessageProtobuf.Message msg, Busi busiServer) {
        //1. get size of group, the receipt of message is the id of group
        String groupCacheKey = PREFIX_GROUP + msg.getHeader().getTenantID() + ":" + groupId;
        RedisCommands<String, String> commands = getConnection().sync();
        long totalMemberCount = commands.zcard(groupCacheKey);
        //1.1 If the size of group is large, to send the message to BusiBridge.
        if(totalMemberCount > Busi.LARGE_GROUP_SIZE &&
                busiServer.getNodeInfo().isPresent() &&
                !busiServer.getNodeInfo().get().isHighPerformance()) {
            try {
                this.transferToNodeProcessingLargeGroup(msg, busiServer);
            }
            catch (BungarusClientException e) {
                e.printStackTrace();
            }
        }
        else {
            //1.2 If the size of group is normal, to broadcast the message to every member in the group
            int i = 0;
            List<String> members = null;
            try {
                members = commands.zrange(groupCacheKey, i, i + size - 1);
                while (members.size() > 0) {
                    //get member's session from session container
                    List<Object> sessionsList = SessionManagement.getInstance().retrieveM(members,
                            String.valueOf(msg.getHeader().getTenantID()));
                    logger.debug("sessionsList -> " + sessionsList);

                    //Persist offline message at first
                    OfflineMessagePersist offlineMessagePersist = OfflineMessagePersist.getInstance(busiServer.getMongoTemplate4FailedRetry());
                    offlineMessagePersist.doBulkBusiness(msg, members, sessionsList);

                    //send status message to access node from which the user connected
                    sessionsList.stream().forEach(sessions -> {
                        Set userSession = (Set) sessions;
                        sendToAccess(userSession, msg, busiServer);
                        userSession.clear();
                    });


                    sessionsList.clear();
                    sessionsList = null;
                    members.clear();
                    members = null;

                    i = i + size;
                    members = commands.zrange(groupCacheKey, i, i + size - 1);
                }
            }
            catch (Exception e) {
                e.printStackTrace();
            }
            finally {
                if (null != members) {
                    members.clear();
                    members = null;
                }
            }
        }
    }

    private void sendToAccess(Set userSession, MessageProtobuf.Message msg, Busi busiServer) {
        userSession.stream().forEach(s -> {
            //value pattern: {userId},{sessionId},{ipport}
            logger.debug("user session string -> " + userSession);
            String[] values = StringUtils.tokenizeToStringArray(s.toString(), SessionManagement.VALUE_SEPERATOR);
            String reciptId = values[0];
            String sessionId = values[1];
            String ipport = values[2];
            if(!reciptId.equals(msg.getHeader().getFrom())) {//Avoid sending message to itself
                List<InnerClient> accessClients = busiServer.getInnerClientList().get(Busi.CLUSTER_TAG_ACCESS);
                for(InnerClient accessClient: accessClients) {
                    if(ipport.equals(accessClient.getSrvIp() + ":" + accessClient.getSrvPort())) {
                        try {
                            MessageProtobuf.Message mmsg = MessageBuilder.buildGroupIMMessage(sessionId, reciptId,
                                    Busi.CLUSTER_TAG_ACCESS, msg);
                            accessClient.sendMessage(mmsg, Busi.CLUSTER_TAG_ACCESS);
                        }
                        catch (BungarusClientException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        });
    }
}
